package no.ffi.gismoidm.pubsub;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import no.ffi.gismoidm.idp.Config;

/**
 * This class contains the necessary info for forwarding decisions. The
 * methods for maintenance and dissemination of forwarding info is included
 * in the class
 * @author anders
 */
public class ForwardingTable implements Runnable {
    private class TableEntry {
        /*
         * One table entry for each MR that has announced subscription interests.
         * Should we have a separate table for neighbors, or use the same table
         * (possibly with null subscriptions). Using one single table is easier-
         * and we could generate table entries based on configuration data.
         * During connection establishment, the MRid of the neighbor should
         * be the same as subjectName from the authentication.
         * 
         * The Forwarding table also works as a neighbor (link) table. The
         * rule is that every MR authenticates with its neighbors and sends
         * an sggSub (even an empty if necessary). The neighbor is detected
         * if the message router id in the NextHop and AggSub variables are
         * the same.
         */
        AggregateSubscription aggSub; // AggSub of this node
        MessageRouter nextHop;
        Long timestampAdded;// Needed to detect stale routes
        boolean replaceIfNewer(AggregateSubscription as, MessageRouter sender) {
            if (!as.mrID.equals(aggSub.mrID)) return false; // Not his entry
            if (as.timestamp > aggSub.timestamp) { // More recent entry
                aggSub = as;
                nextHop = sender;
                timestampAdded = System.currentTimeMillis();
                System.out.println("FWT entry replace by a more recent value");
                return true;
            } else {
                System.out.println("Received aggSub on " + as.mrID + ", but was old and thus discarded");
                return false;
            }
        }
        public String toString() {
            return "aggSub.mrID=" + aggSub.mrID + ", nextHop=" + nextHop.messageNeighborID + ", ts=" + aggSub.timestamp;
        }
        /*
         * Aggregate subscriptions are only sent to the neighbors. This
         * method detects neighbors.
         */
        private boolean isNeighbor() {
            String aggSubID = aggSub.mrID;
            return neighbors.containsKey(aggSubID);
        }
    }
    
    private void dumpFwTable() {
        Iterator<TableEntry> it = forwardTable.iterator();
        while (it.hasNext()) {
            TableEntry te = it.next();
            System.out.println(te);
        }
    }

    private static ForwardingTable instance=null;
    List<TableEntry> forwardTable; // TODO must be threadsafe
    ClientTable clientTable;
    HashMap neighbors;
    Set seenPublications; // duplcate detector
    
    // Ensure a singleton pattern through a private constructor
    private ForwardingTable() {
        forwardTable = new ArrayList();
        clientTable = ClientTable.getInstance();
        neighbors = new HashMap();
        seenPublications = new ExpiringHashSet(1000,2000);
        // Start thread for aggSub dissemination
        new Thread(this).start();
    }
    
    public synchronized static ForwardingTable getInstance() {
        if (instance==null) instance = new ForwardingTable();
        return instance;
    }
    
    // remove all MRs along the route of that neighbor
    public void removeNeighbor(String MRid) {
        Iterator<TableEntry> it = forwardTable.iterator();
        while (it.hasNext()) {
            TableEntry te = it.next();
            if (te.nextHop.equals(MRid)) {
                System.out.println("Deleting MR entry for " + MRid);
                it.remove();
            }
        }
        dumpFwTable();
    }
    
    public void setNeighborEntry(String mrId, MessageRouter mr) { neighbors.put(mrId,mr); }
    
    /*
     * This method processes an incoming aggregate subscription, received
     * from another MR. If its timestamp is later than our entry for that
     * MR, we will replace and forward to all neighbors. Otherwise discard it.
     */
    public void aggregateSubscriptionReceived(AggregateSubscription aggSub, MessageRouter sender) {
        System.out.println("Received aggregate subscription about " + aggSub.mrID + " from " + sender.messageNeighborID);
        System.out.println("it has timestamp " + aggSub.timestamp);
        System.out.println("Size of forwardtable is " + forwardTable.size());
        //long ts = aggSub.timestamp;
        String mrId = aggSub.mrID;
        if (clientTable.getMyId().equals(aggSub.mrID)) {
            System.out.println("This aggSub is my own, and will not be stored");
            return;
        }
        // Look up MR id in table
        boolean known = false;
        Iterator<TableEntry> it = forwardTable.iterator();
        while (it.hasNext()) {
            TableEntry te = it.next();
            if (aggSub.mrID.equals(te.aggSub.mrID)) {
                known = true;
                if (te.replaceIfNewer(aggSub, sender)) {
                    // Send this announcement to all neighbors, except the sender
                    broadcastAggSub(aggSub,sender);
                    dumpFwTable();
                    return;
                }
            }
        } 
        if (!known) {
            // This MRid is not in table, create a new entry
            TableEntry te = new TableEntry();
            te.aggSub = aggSub;
            te.nextHop = sender;
            te.timestampAdded = System.currentTimeMillis();
            System.out.println("Legger til aggSub i FT:" + te);
            forwardTable.add(te);
            // Send this announcement to all neighbors, except the sender
            broadcastAggSub(aggSub,sender);
        }
        dumpFwTable();
    }
    
    /*
     * Send an aggSub to all neighbors, except the neighbor that sent it to us,
     * as a part of a flooding scheme. This procedure should be run on scheduled intervals
     */
    private void broadcastAggSub(AggregateSubscription aggSub, MessageRouter sender) {
        if (aggSub.isMarker()) {
            System.out.println("This aggSub is a marker, and will not be dissemminated");
            return;
        }
        System.out.println("Sending aggsub to neighbors:");
        // iterate over the neighbor table
        Collection c = neighbors.values();
        Iterator<MessageNeighbor> it = c.iterator();
        while (it.hasNext()) {
            MessageNeighbor nb = it.next();
            System.out.println("..sending to " + nb.messageNeighborID);
            nb.sendMessageTo(aggSub);
        }
    }
    
    public void newLocalAggSub(AggregateSubscription aggsub) {
        broadcastAggSub(aggsub, null);
    }
    
    
    /*
     * When a publication is received, it should be delivered to local subscribers
     * and to other MR, based on the subscription info. We must iterate over
     * the local client table and the forward table. Only one copy of the
     * publication is sent to a neighbor, even if there are several MRs along
     * that route. Don't send back to sending neighbor, of course. 
     */
    public void publicationReceived(Publication pub, MessageNeighbor sender) {
        if (seenPublications.contains(pub.getTimeCreated())) {
            System.out.println("Discarding a duplicate publication");
            return;
        } else {
            seenPublications.add(pub.getTimeCreated());
        }
        
        
        List<MessageClient> cl = clientTable.matchPublication(pub);
        // Now cl holds the list of clients that should receive
        // the publication. Pass the list off to someone who can handle
        // that transport asynchronously
        System.out.println("Publication received, " + pub.getTimeCreated());
        Iterator<MessageClient> it = cl.iterator();
        while (it.hasNext()) {
            MessageClient c1 = it.next();
            System.out.println("Sending publication to client " + c1.messageNeighborID);
            c1.sendMessageTo(pub);
        }
        
        // Now find out who will receive this publication. Identify
        // the first hop on the path to them, and send no more than one
        // copy to each neighbor
        
        HashSet<MessageNeighbor> nb = new HashSet();
        Iterator<TableEntry> in = forwardTable.iterator();
        while (in.hasNext()) {
            TableEntry te = in.next();
            AggregateSubscription as = te.aggSub;
            MessageNeighbor next = te.nextHop;
            // Now check if message should be forwarded to this one
            // Unless that neighbor is already in the forwarding seg
            if (next == sender) continue; // Don't send back to sender
            if (nb.contains(next)) continue; // Neighbor already marked
            if (as.forwardMatch(pub)) nb.add(next);
        }
        
        // Now iterate over the hashset and send the publication to
        // all neighbors in the set
        Iterator<MessageNeighbor> im = nb.iterator();
        while (im.hasNext()) {
            MessageNeighbor n2 = im.next();
            System.out.println("Sending publication to MR " + n2.messageNeighborID);
            n2.sendMessageTo(pub);
        }
    }
    
    // Thread responsible for regular dissemmination of aggSub object to neighbors
    public void run() {
        long dissInterval = Long.parseLong(Config.getProperty("AGGSUB_BROADCAST_INTERVAL", "60"))*1000;
        while (true) {
            try { Thread.sleep(dissInterval); } // TODO pick up from config file
            catch (Exception e) {}
            newLocalAggSub(clientTable.generateLocalAggSub());
            
            // Now remove stale entries which have not been refreshed for
            // three sleep periods
            Iterator<TableEntry> it = forwardTable.iterator();
            while (it.hasNext()) {
                TableEntry te = it.next();
                // check age of aggSub entry. If too old, then delete it
                if (System.currentTimeMillis()-te.timestampAdded > 3*dissInterval) {
                    it.remove();
                    System.out.println("Removed forwarding entry for " + te.aggSub.mrID);
                }
            }
            
        }
        
    }
}
